package main

import (
	"context"
	"sync"
	"sync/atomic"
	"time"

	"gitee.com/chenhonghua/ginorigin/env"
	"gitee.com/chenhonghua/ginorigin/log"
	"github.com/bytedance/gopkg/util/gopool"
	redis "github.com/go-redis/redis/v8"
)

var (
	contextFrom                   = context.Background()
	contextTo                     = context.Background()
	REDIS_FROM_HOST               = env.StringVal("REDIS_FROM_HOST", "127.0.0.1:6379")
	REDIS_TO_HOST                 = env.StringVal("REDIS_TO_HOST", "127.0.0.1:6379")
	clientFrom      *redis.Client = redis.NewClient(&redis.Options{
		Addr:     REDIS_FROM_HOST,
		Password: env.StringVal("REDIS_FROM_PWD", ""),
		DB:       env.IntVal("REDIS_FROM_DB", 0),
	})
	clientTo *redis.Client = redis.NewClient(&redis.Options{
		Addr:     REDIS_TO_HOST,
		Password: env.StringVal("REDIS_TO_PWD", ""),
		DB:       env.IntVal("REDIS_TO_DB", 0),
	})
	keyTypeMap     sync.Map = sync.Map{}
	SYNC_TIMES_MAX int64    = env.Int64Val("SYNC_TIMES_MAX", 0) // 0表示
	sync_times     int64    = 0
)

func sync_key(key string) (int64, int64) {
	var dt string
	if dtv, ok := keyTypeMap.Load(key); !ok {
		var e error
		if dt, e = clientFrom.Type(contextFrom, key).Result(); e != nil {
			log.Fatalf("获取key[%s]类型失败: %s", key, e)
		}
		keyTypeMap.Store(key, dt)
	} else {
		dt = dtv.(string)
	}
	switch dt {
	case "string":
		return sync_key_string(key), 1
	case "list":
		return sync_key_list(key)
	case "hash":
		return sync_key_hash(key)
	default:
		log.Errorf("未知的数据类型: key=%s, datatype=%s", key, dt)
		return 0, 0
	}
	// log.Debugf("key=%s, datatype=%s, size=%d", key, dt, size)
}

func sync_key_string(key string) int64 {
	if vf, e := clientFrom.Get(contextFrom, key).Result(); e != nil {
		log.Fatalf("获取key[%s][string][from]值失败: %s", key, e)
	} else if len(vf) == 0 {
		return 0
	} else if vt, e := clientTo.Get(contextTo, key).Result(); e != nil || vf != vt {
		clientTo.Set(contextTo, key, vf, 0)
		log.Infof("update[string]: key=%s", key)
		return 1
	}
	return 0
}

func sync_key_list(key string) (int64, int64) {
	size := clientFrom.LLen(contextFrom, key).Val()
	if size == 0 {
		return 0, 0
	}
	arrFrom, e := clientFrom.LRange(contextFrom, key, 0, size).Result()
	if e != nil {
		log.Fatalf("获取key[%s][list][from]失败: %s", key, e)
	}
	sizeTo := clientTo.LLen(contextTo, key).Val()
	if sizeTo > 0 {
		isSame := true
		if size != sizeTo {
			isSame = false
		} else { // 当长度一样时，对list类型数据进行对比
			if arrTo, e := clientTo.LRange(contextTo, key, 0, sizeTo).Result(); e == nil {
				for i, from := range arrFrom {
					if from != arrTo[i] {
						isSame = false
						break
					}
				}
			}
		}
		if isSame { // 目标redis中数据与源数据相同，无需进行处理
			log.Debugf("list is same: key=%s, len=%d", key, size)
			return 0, size
		}
	}
	clientTo.Del(contextTo, key)
	arrFromInterface := []interface{}{}
	for _, f := range arrFrom {
		arrFromInterface = append(arrFromInterface, f)
	}
	clientTo.RPush(contextTo, key, arrFromInterface...)
	log.Infof("update[list]: key=%s, len=%d->%d", key, sizeTo, size)
	return size, size
}

func sync_key_hash(key string) (int64, int64) {
	size := clientFrom.HLen(contextFrom, key).Val()
	if size == 0 {
		return 0, 0
	}
	keyValuesFrom, e := clientFrom.HGetAll(contextFrom, key).Result()
	if e != nil {
		log.Fatalf("获取key[%s][hash]失败: %s", key, e)
	}
	toSize := clientTo.HLen(contextTo, key).Val()
	if size != toSize {
		clientTo.Del(contextTo, key)
	} else {
		keyValuesTo, e := clientFrom.HGetAll(contextFrom, key).Result()
		if e != nil {
			clientTo.Del(contextTo, key)
		} else {
			isSame := true
			for k, v := range keyValuesFrom {
				if v != keyValuesTo[k] {
					isSame = false
					clientTo.Del(contextTo, key)
					break
				}
			}
			if isSame { // 完全一样，不进行处理
				return 0, size
			}
		}
	}
	clientTo.HSet(contextTo, key, keyValuesFrom)
	log.Infof("update[hash]: key=%s, len=%d", key, size)
	return size, size
}

func doSync() {
	log.Info("------------------------------------------------------------------------------------------")
	st := time.Now().Unix()
	keys, e := clientFrom.Keys(contextFrom, "*").Result()
	if e != nil {
		log.Fatalf("获取[from]所有key失败: %s", e)
	}
	changedCount := int64(0)
	completeCount := int64(0)
	poolCapacity := env.Uint64Val("POOL_CAPACITY", 1)
	if poolCapacity > 1 {
		gopool.SetCap(int32(poolCapacity))
	}
	var wg sync.WaitGroup
	for _, key := range keys {
		if poolCapacity == 0 || poolCapacity == 1 {
			change, complete := sync_key(key)
			changedCount += change
			completeCount += complete
			continue
		}
		wg.Add(1)
		gopool.Go(func() {
			defer wg.Done()
			change, complete := sync_key(key)
			atomic.AddInt64(&changedCount, change)
			atomic.AddInt64(&completeCount, complete)
		})
	}
	wg.Wait()
	et := time.Now().Unix()
	log.Infof("全部键值[%d/%d/%d]处理完成, 耗时: %d秒", changedCount, completeCount, len(keys), et-st)
	if changedCount == 0 && SYNC_TIMES_MAX > 0 {
		sync_times++
	}
	log.Info("------------------------------------------------------------------------------------------")
}

// 工具：redis同步。
// 将源redis中的数据同步到目标redis中，目标redis中的数据不用被删除，只会覆盖相同的键。
// 初次执行由于需要进行数据迁移，所以可能耗时较长，但后续的匹配同步操作，执行速度会非常快。
func main() {
	log.SetLogLevel(env.StringVal("LOG_LEVEL", "info"))
	if REDIS_FROM_HOST == REDIS_TO_HOST {
		log.Fatalf("源和目标redis地址不能相同")
	}
	if e := clientFrom.Ping(contextFrom).Err(); e != nil {
		log.Fatalf("REDIS[from][%s]连接失败: %s", REDIS_FROM_HOST, e)
	}
	log.Infof("REDIS[from][%s]连接成功", REDIS_FROM_HOST)
	if e := clientTo.Ping(contextTo).Err(); e != nil {
		log.Fatalf("REDIS[to][%s]连接失败: %s", REDIS_TO_HOST, e)
	}
	log.Infof("REDIS[to][%s]连接成功", REDIS_TO_HOST)
	for {
		doSync()
		if SYNC_TIMES_MAX > 0 && sync_times >= SYNC_TIMES_MAX {
			break
		}
		time.Sleep(1e9) // 休眠1秒
	}
}
